package com.zjl.学习.第07章_高级.C_分区步骤;

import com.zjl.springbatch.util.OutColour;
import com.zjl.学习.第07章_高级.User;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepScope;
import org.springframework.batch.core.launch.support.RunIdIncrementer;
import org.springframework.batch.core.partition.PartitionHandler;
import org.springframework.batch.core.partition.support.TaskExecutorPartitionHandler;
import org.springframework.batch.core.step.tasklet.TaskletStep;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.file.FlatFileItemReader;
import org.springframework.batch.item.file.builder.FlatFileItemReaderBuilder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
import org.springframework.core.io.Resource;
import org.springframework.core.task.SimpleAsyncTaskExecutor;

import java.nio.charset.Charset;
import java.util.List;
/**
 * 分区：有划分，区分意思，在SpringBatch 分区步骤讲的是给执行步骤区分上下级。
 *      上级： 主步骤(Master Step)
 *      下级： 从步骤–工作步骤(Work Step)
 * 主步骤是领导，不用干活，负责管理从步骤，从步骤是下属，必须干活。
 *      一个主步骤下辖管理多个从步骤。
 * 注意： 从步骤，不管多小，它也一个完整的Spring Batch 步骤，负责各自的读入、处理、写入等。
 *
 *
 * 分区步骤一般用于海量数据的处理上，其采用是分治思想。
 * 主步骤将大的数据划分多个小的数据集，然后开启多个从步骤，
 * 要求每个从步骤负责一个数据集。当所有从步骤处理结束，整作业流程才算结束。
 *
 *
 * 1>文件分区器：userPartitioner()， 分别加载5个文件进入到程序
 * 2>文件分区处理器：userPartitionHandler() ，指定要分几个区，由谁来处理
 * 3>分区从步骤：workStep() 指定读逻辑与写逻辑
 * 4>分区文件读取：flatItemReader()，需要传入Resource对象，这个对象在userPartitioner()已经标记为file
 * 5>分区主步骤：masterStep() ，指定分区名称与分区器，指定分区处理器
 *
 */
@SpringBootApplication
@EnableBatchProcessing
public class C0_分区步骤 {
    @Autowired
    private JobBuilderFactory jobBuilderFactory;
    @Autowired
    private StepBuilderFactory stepBuilderFactory;

    //设置  每个分区文件读取
    @Bean
    @StepScope
    public FlatFileItemReader<User> flatItemReader(@Value("#{stepExecutionContext['file']}") Resource resource/*获取 step中公共变量的指定数据*/){
        return new FlatFileItemReaderBuilder<User>()
                .name("userItemReader")
                .resource(resource)
                .delimited().delimiter("#")
                .names("id", "name", "age")
                .targetType(User.class)
                .build();
    }

    @Bean
    public ItemWriter<User> itemWriter(){
        return new ItemWriter<User>() {
            @Override
            public void write(List<? extends User> items) throws Exception {
                OutColour.out.printlnRed(items);
            }
        };
    }


    //文件分区器-设置分区规则
    @Bean
    public MyPartitioner myPartitioner(){
        return new MyPartitioner();
    }

    //文件分区处理器-处理分区  设置分区信息
    @Bean
    public PartitionHandler userPartitionHandler() {
        TaskExecutorPartitionHandler handler = new TaskExecutorPartitionHandler();
        handler.setGridSize(5);//表示有5个分区
        handler.setTaskExecutor(new SimpleAsyncTaskExecutor());//表示各分区异步处理
        handler.setStep(workStep());//放入分步骤的  步骤
        try {
            handler.afterPropertiesSet();//进行 步骤判空
        } catch (Exception e) {
            e.printStackTrace();
        }
        return handler;
    }

    //每个从分区操作步骤   配置分区的 shep 步骤
    @Bean
    public Step workStep() {
        TaskletStep workStep =  stepBuilderFactory.get("workStep")
                .<User, User>chunk(2)
                .reader(flatItemReader(null))//放入读
                .writer(itemWriter())//放入写
                .build();
        return workStep;
    }

    //主分区操作步骤
    @Bean
    public Step masterStep() {
        return stepBuilderFactory.get("masterStep")
                //放入分步骤的 名字，表示对谁进行步骤的分区操作    放入分区规则
                .partitioner(workStep().getName(), myPartitioner())
                .partitionHandler(userPartitionHandler())//放入分区器
                .build();
    }


    @Bean
    public Job partJob(){
        return jobBuilderFactory.get("part-job")
                .start(masterStep())

                .incrementer(new RunIdIncrementer())

                .build();
    }
    public static void main(String[] args) {
        System.setProperty("spring.profiles.active","01");//  同理 --spring.profiles.active=xxx
        SpringApplication.run(C0_分区步骤.class, args);
        System.out.println("***********启动完毕**当前编码:" + Charset.defaultCharset().name() + "****************");

    }
}

